package zhang.app.dws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import zhang.app.BaseAppV1;
import zhang.bean.TestCourseTestPTestPaperBean;
import zhang.common.Constant;
import zhang.util.AtguiguUtil;
import zhang.util.FlinkSinkUtil;

import java.math.BigInteger;
import java.time.Duration;

public class DwsTestCourseTestTestPaperWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new DwsTestCourseTestTestPaperWindow().init(3016,2,"DwsTestCourseTestTestPaperWindow", Constant.TOPIC_DWD_TEST_TEST);

    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. 封装到bean
        SingleOutputStreamOperator<TestCourseTestPTestPaperBean> beanStream = pasebean(stream);


        // 2. 开窗聚合
        SingleOutputStreamOperator<TestCourseTestPTestPaperBean> resultStream = windowAndAgg(beanStream);

//            //3. 写出
        writeToDoris(resultStream);
    }

    private void writeToDoris(SingleOutputStreamOperator<TestCourseTestPTestPaperBean> resultStream) {
        resultStream.map(bean -> {
            SerializeConfig config = new SerializeConfig();
            config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
            String json = JSON.toJSONString(bean, config);
            System.out.println(json);
            return json;
        })
                .addSink(FlinkSinkUtil.getDorisSink("dws_db.dws_test_course_test_test_paper_view_window"));
    }

    private SingleOutputStreamOperator<TestCourseTestPTestPaperBean>
    windowAndAgg(SingleOutputStreamOperator<TestCourseTestPTestPaperBean> beanStream) {
      return   beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.
                <TestCourseTestPTestPaperBean>forBoundedOutOfOrderness(Duration.ofSeconds(60))
        .withTimestampAssigner((bean,ts)->bean.getTs()*1000))

                .keyBy(bean->bean.getCourseId()+"_"+bean.getPaperId()+"_"+bean.getPointRank())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TestCourseTestPTestPaperBean>() {

                            @Override
                            public TestCourseTestPTestPaperBean reduce(TestCourseTestPTestPaperBean bean1,
                                                                       TestCourseTestPTestPaperBean bean2) throws Exception {

                               bean1.setUserNum(bean1.getUserNum()+bean2.getUserNum());
                               bean1.setScore(bean1.getScore().add(bean2.getScore()));
                               bean1.setDurationSec(bean1.getDurationSec().add(bean2.getDurationSec()));
                               return bean1;
                            }
                        },
                        new ProcessWindowFunction<TestCourseTestPTestPaperBean, TestCourseTestPTestPaperBean, String, TimeWindow>() {


                            @Override
                            public void process(String s,
                                                Context context,
                                                Iterable<TestCourseTestPTestPaperBean> elements,
                                                Collector<TestCourseTestPTestPaperBean> out) throws Exception {

                                TestCourseTestPTestPaperBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));
                                bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis()));

                                out.collect(bean);
                            }
                        }
                        );




    }

    private SingleOutputStreamOperator<TestCourseTestPTestPaperBean> pasebean(DataStreamSource<String> stream) {

      return   stream.map(json-> {
            JSONObject object = JSON.parseObject(json);
          return   new TestCourseTestPTestPaperBean(
                "",
                    "",
                    object.getBigInteger("paper_id"),
                    object.getBigInteger("course_id"),
                    object.getString("point_rank"),
                    object.getString("user_id"),
                    1l,
                    object.getBigDecimal("score"),
                    object.getBigInteger("duration_sec"),
                    "",
                    object.getLong("ts")
            );

        });
    }
}


